#include "../lib_include/TcpConnection.h"
#include "../lib_include/Logger.h"
#include "../lib_include/WeakCallback.h"
#include "../lib_include/Channel.h"
#include "../lib_include/EventLoop.h"
#include "../lib_include/Socket.h"
#include "../lib_include/SocketsOps.h"

#include <errno.h>

using namespace netlib;
using namespace netlib::net;

// 默认连接到来的回调函数
void netlib::net::defaultConnectionCallback(const TcpConnectionPtr& conn)
{
    LOG_TRACE << conn->localAddress().toIpPort() << " -> "
                << conn->peerAddress().toIpPort() << " is "
                << (conn->connected() ? "UP" : "DOWN");
}

// 默认消息到来的回调函数
void netlib::net::defaultMessageCallback(const TcpConnectionPtr&,
                                        Buffer* buf,
                                        Timestamp)
{ buf->retrieveAll(); }

TcpConnection::TcpConnection(EventLoop* loop,
                             const string& nameArg,
                             int sockfd,
                             const InetAddress& localAddr,
                             const InetAddress& peerAddr)
  : loop_(CHECK_NOTNULL(loop)), // CHECK_NOTNULL检查loop是否为空指针
    name_(nameArg),
    state_(kConnecting), // 初始化状态为建立连接中
    reading_(true),
    socket_(new Socket(sockfd)),
    channel_(new Channel(loop, sockfd)),
    localAddr_(localAddr),
    peerAddr_(peerAddr),
    highWaterMark_(64*1024*1024)
{
    channel_->setReadCallback( // 设置各个事件的处理函数
        std::bind(&TcpConnection::handleRead, this, _1));
    channel_->setWriteCallback(
        std::bind(&TcpConnection::handleWrite, this));
    channel_->setCloseCallback(
        std::bind(&TcpConnection::handleClose, this));
    channel_->setErrorCallback(
        std::bind(&TcpConnection::handleError, this));
    LOG_DEBUG << "TcpConnection::ctor[" <<  name_ << "] at " << this
                << " fd=" << sockfd;
    socket_->setKeepAlive(true); // 开启内核的心跳
}


TcpConnection::~TcpConnection()
{
    LOG_DEBUG << "TcpConnection::dtor[" <<  name_ << "] at " << this
                << " fd=" << channel_->fd()
                << " state=" << stateToString();
    assert(state_ == kDisconnected);
}

// 获取TCP相关信息
bool TcpConnection::getTcpInfo(struct tcp_info* tcpi) const
{ return socket_->getTcpInfo(tcpi); }


string TcpConnection::getTcpInfoString() const // 获取TCP相关信息并将其转化成字符串
{
    char buf[1024];
    buf[0] = '\0';
    socket_->getTcpInfoString(buf, sizeof buf);
    return buf;
}

// 发送数据，传入void*和len
void TcpConnection::send(const void* data, int len)
{ send(StringPiece(static_cast<const char*>(data), len)); }

// 发送数据，传入StringPiece&
void TcpConnection::send(const StringPiece& message)
{
    if (state_ == kConnected)
    {
        if (loop_->isInLoopThread()) // 如果在IO线程中，则直接发送
            sendInLoop(message);
        else // 添加到IO线程附加任务列表
        {
            void (TcpConnection::*fp)(const StringPiece& message) = &TcpConnection::sendInLoop;
            loop_->runInLoop(std::bind(fp, this, message.as_string()));
        }
    }
}

// 发送数据，传入Buffer*
void TcpConnection::send(Buffer* buf)
{
    if (state_ == kConnected)
    {
        if (loop_->isInLoopThread()) // 在IO线程中直接发送
        {
            sendInLoop(buf->peek(), buf->readableBytes());
            buf->retrieveAll();
        }
        else // 添加到IO线程附加任务列表
        {
            void (TcpConnection::*fp)(const StringPiece& message) = &TcpConnection::sendInLoop;
            loop_->runInLoop(std::bind(fp, this, buf->retrieveAllAsString()));
        }
    }
}

// 发送数据，传入StringPiece&
void TcpConnection::sendInLoop(const StringPiece& message)
{ sendInLoop(message.data(), message.size()); }

// 发送数据，执行发送行为
void TcpConnection::sendInLoop(const void* data, size_t len)
{
    loop_->assertInLoopThread();
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool faultError = false;
    if (state_ == kDisconnected) // 连接已经关闭，放弃发送
    {
        LOG_WARN << "disconnected, give up writing";
        return;
    }
    if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
    { // 当前通道没有监听可写事件且写缓冲区为空，直接发送，不用添加到缓冲区
        nwrote = sockets::write(channel_->fd(), data, len); // 向连接写入数据
        if (nwrote >= 0)
        {
            remaining = len - nwrote;
            if (remaining == 0 && writeCompleteCallback_) // 全部发送完毕，调用低水位
                loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
        }
        else // 写入出错
        {
            nwrote = 0;
            if (errno != EWOULDBLOCK)
            {
                LOG_SYSERR << "TcpConnection::sendInLoop";
                if (errno == EPIPE || errno == ECONNRESET) // FIXME: any others?
                    faultError = true; // 设置写入错误状态
            }
        }
    }

    assert(remaining <= len);
    if (!faultError && remaining > 0)
    {
        size_t oldLen = outputBuffer_.readableBytes();

        // 缓冲区已满，调用高水位回调函数
        if (oldLen + remaining >= highWaterMark_
            && oldLen < highWaterMark_
            && highWaterMarkCallback_)
            loop_->queueInLoop(std::bind(highWaterMarkCallback_,
                                            shared_from_this(), oldLen + remaining));
        
        // 将数据添加到缓冲区中
        outputBuffer_.append(static_cast<const char*>(data)+nwrote, remaining);
        if (!channel_->isWriting())
            channel_->enableWriting(); // 注册监听通道的可写事件
    }
}

// 关闭连接，shutdownInLoop的封装，Buffer中发完再关闭
void TcpConnection::shutdown()
{
    if (state_ == kConnected)
    {
        setState(kDisconnecting); // 设置连接状态为正在销毁中
        loop_->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));
    }
}

// 关闭连接
void TcpConnection::shutdownInLoop()
{
    loop_->assertInLoopThread();
    if (!channel_->isWriting()) // 如果该该通道还在关注POLLOUT事件，则表示正在写入
        socket_->shutdownWrite(); // 不处于正在写状态，则关闭写端
}

// 超时关闭连接，forceCloseInLoop的封装
void TcpConnection::forceClose()
{
    if (state_ == kConnected || state_ == kDisconnecting)
    {
        setState(kDisconnecting);
        loop_->queueInLoop(std::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
    }
}

// 创建一个定时器，超时关闭连接
void TcpConnection::forceCloseWithDelay(double seconds)
{
    if (state_ == kConnected || state_ == kDisconnecting)
    {
        setState(kDisconnecting);
        loop_->runAfter(seconds, // 一次性定时器
                        makeWeakCallback(shared_from_this(),
                        &TcpConnection::forceClose));
    }
}

// 超时关闭连接
void TcpConnection::forceCloseInLoop()
{
    loop_->assertInLoopThread();
    if (state_ == kConnected || state_ == kDisconnecting)
        handleClose(); // 关闭事件回调函数
}

const char* TcpConnection::stateToString() const // 将连接状态转成字符串返回
{
    switch (state_)
    {
        case kDisconnected:
            return "kDisconnected";
        case kConnecting:
            return "kConnecting";
        case kConnected:
            return "kConnected";
        case kDisconnecting:
            return "kDisconnecting";
        default:
            return "unknown state";
    }
}

// 禁用Nagle算法，可以避免连续发包出现延迟，对于低延迟服务很重要
void TcpConnection::setTcpNoDelay(bool on)
{ socket_->setTcpNoDelay(on); }

// 将连接注册事件表的可读事件，startReadInLoop的封装
void TcpConnection::startRead()
{ loop_->runInLoop(std::bind(&TcpConnection::startReadInLoop, this)); }

void TcpConnection::startReadInLoop() // 将连接注册事件表的可读事件
{
    loop_->assertInLoopThread();
    if (!reading_ || !channel_->isReading())
    {
        channel_->enableReading();
        reading_ = true;
    }
}

// 撤销连接的监听可读事件，stopReadInLoop的封装
void TcpConnection::stopRead()
{ loop_->runInLoop(std::bind(&TcpConnection::stopReadInLoop, this)); }


void TcpConnection::stopReadInLoop() // 撤销连接的监听可读事件
{
    loop_->assertInLoopThread();
    if (reading_ || channel_->isReading())
    {
        channel_->disableReading();
        reading_ = false;
    }
}

void TcpConnection::connectEstablished() // 将连接插入到事件表中进行关注
{
    loop_->assertInLoopThread();
    assert(state_ == kConnecting); // 断言处于连接建立中
    setState(kConnected); // 设置连接状态

    // 连接建立成功后，设置副本weak_ptr和标志位，每次通道处理事件的时候提升，计数加一
    // 防止调用关闭连接时，从列表中删除连接的时候因为shared_ptr计数为0，提前析构导致程序崩溃
    channel_->tie(shared_from_this());

    channel_->enableReading(); // 向poll/epoll注册可读事件

    connectionCallback_(shared_from_this());
}

void TcpConnection::connectDestroyed() // 销毁连接
{
    loop_->assertInLoopThread();
    if (state_ == kConnected) // 检查连接状态
    {
        setState(kDisconnected); // 设置连接状态
        channel_->disableAll();
        connectionCallback_(shared_from_this()); // shared_from_this保证计数加一
    }
    channel_->remove(); // 在事件表中移除通道
}

void TcpConnection::handleRead(Timestamp receiveTime) // 可读事件处理函数
{
    loop_->assertInLoopThread(); // 断言创建者线程
    int savedErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
    if (n > 0)
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    else if (n == 0) // read==0，即关闭连接
        handleClose();
    else // read<0,出错
    {
        errno = savedErrno;
        LOG_SYSERR << "TcpConnection::handleRead";
        handleError();
    }
}

void TcpConnection::handleWrite() // 可写事件处理函数
{
    loop_->assertInLoopThread();
    if (channel_->isWriting())
    {
        ssize_t n = sockets::write(channel_->fd(),
                                outputBuffer_.peek(),
                                outputBuffer_.readableBytes());
        if (n > 0)
        {
            outputBuffer_.retrieve(n);
            if (outputBuffer_.readableBytes() == 0)
            {
                channel_->disableWriting();
                if (writeCompleteCallback_)
                    loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
                if (state_ == kDisconnecting)
                    shutdownInLoop();
            }
        }
        else
            LOG_SYSERR << "TcpConnection::handleWrite";
    }
    else
        LOG_TRACE << "Connection fd = " << channel_->fd() << " is down, no more writing";
}

void TcpConnection::handleClose() // 关闭事件回调函数
{
    loop_->assertInLoopThread();
    LOG_TRACE << "fd = " << channel_->fd() << " state = " << stateToString();
    assert(state_ == kConnected || state_ == kDisconnecting);
    setState(kDisconnected); // 设置连接状态为已销毁
    channel_->disableAll(); // 向poll/epoll撤销所有事件

    // shared_from_this()将当前对象转化成智能指针作参数传入，保证计数器加一
    TcpConnectionPtr guardThis(shared_from_this());
    connectionCallback_(guardThis); // 调用用户的回调函数
    closeCallback_(guardThis);
}

void TcpConnection::handleError() // 错误事件回调函数
{
    int err = sockets::getSocketError(channel_->fd());
    LOG_ERROR << "TcpConnection::handleError [" << name_
                << "] - SO_ERROR = " << err << " " << strerror_tl(err);
}

